package sink;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class SinkToMysql_Press extends RichSinkFunction<ClassObject.PressRes> {
    //声明连接和预编译语句
    Connection connection = null;
    PreparedStatement insertStmt = null;
    PreparedStatement updateStmt1 = null;
    PreparedStatement updateStmt2 = null;
    PreparedStatement replaceStmt = null;
    // open属于生命周期，避免在invoke()中来一条数据就创建一次数据库连接
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection("jdbc:mysql://192.168.68.140:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "root");
        insertStmt = connection.prepareStatement("insert IGNORE into pressrank (rank,name,num) values (?, ?, ?)");
//        updateStmt1 = connection.prepareStatement("update pressrank set name = ? where rank = ?");
//        updateStmt2 = connection.prepareStatement("update pressrank set num = ? where rank = ?");
        replaceStmt = connection.prepareStatement("replace into pressrank (rank,name,num) values (?, ?, ?)");
    }

    // 每来一条数据，调用连接，执行sql
    @Override
    public void invoke(ClassObject.PressRes value, Context context) throws Exception {

        // 直接执行更新语句，如果没有更新就执行插入
        int j=0;
        for (String i : value.getPressnum()) {
            j=j+1;
            String[] array = i.split(",");
            replaceStmt.setInt(1,j);
            replaceStmt.setString(2,array[0]);
            replaceStmt.setDouble(3,Double.valueOf(array[1]));
            replaceStmt.execute();
            //也可以使用update更新
//            updateStmt1.setInt(1, j);
//            updateStmt2.setInt(1, j);
//            updateStmt1.setString(2, array[0]);
//            updateStmt2.setDouble(2, Double.valueOf(array[1]));
//            updateStmt1.execute();
//            updateStmt2.execute();
//            if (updateStmt1.getUpdateCount() == 0 & updateStmt2.getUpdateCount() == 0) {
//                insertStmt.setInt(1, j);
//                insertStmt.setString(2, array[0]);
//                insertStmt.setDouble(3, Double.valueOf(array[1]));
//                insertStmt.execute();
//            }
        }
    }
}
